package p1TopN;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Iterator;
import java.util.Set;
import java.util.TreeMap;

public class topnMapper extends Mapper<LongWritable, Text,Text,AverageTempBean> {
    TreeMap<AverageTempBean,Text> treeMap=new TreeMap<>();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line=value.toString();
        String[] fields=line.split(",");
        String date=fields[0];
        double at=0;
        double atu=0;
        if("".equals(fields[1])){
            at=Double.MIN_EXPONENT;
        }else{
            at=Double.parseDouble(fields[1]);
        }
        if ("".equals(fields[2])){
            atu=Double.MIN_EXPONENT;
        }else {
            atu=Double.parseDouble(fields[2]);
        }
        String city=fields[3];
        String country=fields[4];
        String latitude=fields[5];
        String longitude=fields[6];

        Text valueOfMap=new Text(date);
        AverageTempBean keyOfMap=new AverageTempBean(at,atu,city,country,latitude,longitude);
        treeMap.put(keyOfMap,valueOfMap);

        //目前是按照降序排序的
        if(treeMap.size()>10){
            //.remove(treeMap.lastKey());
            treeMap.remove(treeMap.lastKey());
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        //因为treeMap不能通过迭代器或foreach遍历，所以才会有以下的过程
        Set<AverageTempBean> atBeans=treeMap.keySet();//可以返回key的set集合
        //然后通过迭代器依次取出这些key，然后再通过key查询出treeMap里面的value然后输出
        Iterator<AverageTempBean> iterator = atBeans.iterator();
        while(iterator.hasNext()){
            AverageTempBean atBean=iterator.next();
            Text text=treeMap.get(atBean);
            context.write(text,atBean);           //Mapper的输出中key是日期，value是bean对象
        }
    }
}
